package com.xbai.spark.recall.data

import org.apache.spark.sql.DataFrame

/**
  * @author xbai
  * @Date 2021/1/11
  */
class Perprocessing {

  def createItemScoreByUser(df: DataFrame): DataFrame = {
    val itemTotalTime: DataFrame = df.selectExpr("userId", "musicId", "cast(remainTime as double)", "cast(durationHour as double)")
      .groupBy("userId", "musicId")
      .sum("remainTime")
      .withColumnRenamed("sum(remainTime)", "itemTotalTime")

    val totalTime: DataFrame = df.selectExpr("userId", "musicId", "cast(remainTime as double)", "cast(durationHour as double)")
      .groupBy("userId")
      .sum("remainTime")
      .withColumnRenamed("sum(remainTime)", "totalTime")

    val data: DataFrame = itemTotalTime.join(totalTime, "userId")
      .selectExpr("userId", "musicId as itemId", "itemTotalTime/totalTime as rating")

    data.write.mode("overwrite").saveAsTable("userItemRating")

    data
  }
}
